/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package test

import (
	"fmt"
	"log"
	"math/rand"
	"net"
	"net/rpc"
	"os"
	"sync"
	"time"
)

// Server 包装了raft.ConsensusModule和rpc.Server，并将后者暴露为RPC端点。同时还负责管理Raft服务器的同伴。
// 这样设计的主要目的是简化用于演示的raft.Server代码。 raft.ConsensusModule可以通过*Server与同伴进行通信，而不必关心RPC服务器的具体细节。
type Server struct {
	mu sync.Mutex

	serverID int64
	peerIDs  []int64

	cm       *ConsensusModule
	rpcProxy *RPCProxy

	rpcServer *rpc.Server
	listener  net.Listener

	peerClients map[int64]*rpc.Client

	ready <-chan interface{}
	quit  chan interface{}
	wg    sync.WaitGroup
}

// NewServer 新建一个Server
func NewServer(serverID int64, peerIds []int64, ready <-chan interface{}) *Server {
	s := new(Server)
	s.serverID = serverID
	s.peerIDs = peerIds
	s.peerClients = make(map[int64]*rpc.Client)
	s.ready = ready
	s.quit = make(chan interface{})
	return s
}

// Serve 处理服务
func (s *Server) Serve() {
	s.mu.Lock()
	s.cm = NewConsensusModule(s.serverID, s.peerIDs, s, s.ready)

	// 创建新的RPC服务，并注册一个RPCProxy，将所有的方法调用转发到 n.cm
	s.rpcServer = rpc.NewServer()
	s.rpcProxy = &RPCProxy{cm: s.cm}
	s.rpcServer.RegisterName("ConsensusModule", s.rpcProxy)

	var err error
	s.listener, err = net.Listen("tcp", ":0")
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("[%v] listening at %s", s.serverID, s.listener.Addr())
	s.mu.Unlock()

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()

		for {
			conn, err := s.listener.Accept()
			if err != nil {
				select {
				case <-s.quit:
					return
				default:
					log.Fatal("accept error:", err)
				}
			}
			s.wg.Add(1)
			go func() {
				s.rpcServer.ServeConn(conn)
				s.wg.Done()
			}()
		}
	}()
}

// DisconnectAll 该方法会断开本服务器到同伴服务器之间的所有客户端连接
func (s *Server) DisconnectAll() {
	s.mu.Lock()
	defer s.mu.Unlock()
	for id := range s.peerClients {
		if s.peerClients[id] != nil {
			s.peerClients[id].Close()
			s.peerClients[id] = nil
		}
	}
}

// Shutdown 方法会关闭该服务器，并等待其正确关闭
func (s *Server) Shutdown() {
	s.cm.Stop()
	close(s.quit)
	s.listener.Close()
	s.wg.Wait()
}

// GetListenAddr 获取监听地址
func (s *Server) GetListenAddr() net.Addr {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.listener.Addr()
}

// ConnectToPeer 连接Peer
func (s *Server) ConnectToPeer(peerID int64, addr net.Addr) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.peerClients[peerID] == nil {
		client, err := rpc.Dial(addr.Network(), addr.String())
		if err != nil {
			return err
		}
		s.peerClients[peerID] = client
	}
	return nil
}

// DisconnectPeer 会断开与peerId对应的同伴服务器之间的连接
func (s *Server) DisconnectPeer(peerID int64) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.peerClients[peerID] != nil {
		err := s.peerClients[peerID].Close()
		s.peerClients[peerID] = nil
		return err
	}
	return nil
}

// Call 调用对应方法
func (s *Server) Call(id int64, serviceMethod string, args interface{}, reply interface{}) error {
	s.mu.Lock()
	peer := s.peerClients[id]
	s.mu.Unlock()

	// 如果在关闭（调用client.Close）之后进行方法调用，会返回错误
	if peer == nil {
		return fmt.Errorf("call client %d after it's closed", id)
	}
	return peer.Call(serviceMethod, args, reply)
}

// RPCProxy 是ConsensusModule中的RPC方法的直通代理类型。
//	作用如下：
//	1 - 模拟RPC传输过程中的微小延迟
//	2 - 防止出现 https://github.com/golang/go/issues/19957 所提的问题
//	3 - 当设置RAFT_UNRELIABLE_RPC时，通过刻意延迟某些消息并丢弃其他消息
//		来模拟可能出现的不可靠连接。
type RPCProxy struct {
	cm *ConsensusModule
}

// RequestVote 请求投票信息
func (rpp *RPCProxy) RequestVote(args RequestVoteArgs, reply *RequestVoteReply) error {
	if len(os.Getenv("RAFT_UNRELIABLE_RPC")) > 0 {
		dice := rand.Intn(10)
		if dice == 9 {
			rpp.cm.dlog("drop RequestVote")
			return fmt.Errorf("RPC failed")
		} else if dice == 8 {
			rpp.cm.dlog("delay RequestVote")
			time.Sleep(75 * time.Millisecond)
		}
	} else {
		time.Sleep(time.Duration(1+rand.Intn(5)) * time.Millisecond)
	}
	return rpp.cm.RequestVote(args, reply)
}

// AppendEntries 处理Entries信息
func (rpp *RPCProxy) AppendEntries(args AppendEntriesArgs, reply *AppendEntriesReply) error {
	if len(os.Getenv("RAFT_UNRELIABLE_RPC")) > 0 {
		dice := rand.Intn(10)
		if dice == 9 {
			rpp.cm.dlog("drop AppendEntries")
			return fmt.Errorf("RPC failed")
		} else if dice == 8 {
			rpp.cm.dlog("delay AppendEntries")
			time.Sleep(75 * time.Millisecond)
		}
	} else {
		time.Sleep(time.Duration(1+rand.Intn(5)) * time.Millisecond)
	}
	return rpp.cm.AppendEntries(args, reply)
}
